package com.shujia.mllib

import org.apache.spark.ml.classification.{LogisticRegression, LogisticRegressionModel}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object Demo2Person {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession.builder()
      .master("local")
      .appName("person")
      .config("spark.sql.shuffle.partitions", 2)
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._


    /**
      * 1、特征工程
      * 将原始数据转换成公式可以识别的数据
      *
      *
      * 我们这里已经是被昨晚特征工程的数据了
      *
      */

    //通过spark sql 读取svm格式的数据
    val personDF: DataFrame = spark
      .read
      .format("libsvm")
      .load("spark/data/人体指标.txt")


    personDF.show(false)

    /**
      * 2、切分训练集和测试集
      * 训练集--- 训练模型
      * 测试集-- 测试模型准确率
      *
      * 一般训练集0.7  测试集0.3
      *
      */

    val splitDF: Array[Dataset[Row]] = personDF.randomSplit(Array(0.7, 0.3))

    //训练集
    val trainDF: Dataset[Row] = splitDF(0)

    //测试集
    val testDF: Dataset[Row] = splitDF(1)


    /**
      * 2、选择算法
      *
      * 结果是离散值-- -分类
      *
      * 逻辑回归
      *
      */

    //构建算法执行参数
    val logisticRegression: LogisticRegression = new LogisticRegression()
      .setMaxIter(10) //最大迭代次数
      .setFitIntercept(true) //是否有截距


    /**
      * 3、训练模型
      * 将训练集带入算法训练模型 --  确定k和b
      *
      * 模型包含公式和参数
      *
      */
    val model: LogisticRegressionModel = logisticRegression.fit(trainDF)


    /**
      * 4、模型评估
      * 使用模型预测测试集的数据，判断和原始标记是否一直，计算准确率
      *
      */
    val frame: DataFrame = model.transform(testDF)

    /**
      *
      * 计算准确率
      */

    val result: DataFrame = frame.select(sum(when($"label" === $"prediction", 1).otherwise(0)) / count($"label"))

    result.show()

    /**
      * 如果准确率还可以就保存模型
      * 保存再hdfs
      *
      */

    model.save("spark/data/model")


  }

}
